package com.yanggu.bigdata.realtime.utils

import cn.hutool.core.util.StrUtil
import com.yanggu.bigdata.realtime.bean.{ProductStats, TransientSink}
import com.yanggu.bigdata.realtime.common.GmallConfig
import org.apache.flink.connector.jdbc.{JdbcConnectionOptions, JdbcExecutionOptions, JdbcSink, JdbcStatementBuilder}

import java.sql.PreparedStatement
import java.util
import scala.util.control.Breaks.{break, breakable}

/**
 * clickhouse的工具类
 */
object ClickHouseUtil {

  /**
   *
   * @param sql sql的形式需要是INSET INTO XXX(XX, XX) VALUES(?, ?) 需要实体类中的字段定义的顺序和SQL中字段的顺序一致
   * @tparam T
   * @return
   */
  def getSinkFunction[T](sql: String) = {

    //参数赋值
    val jdbcStatementBuilder = new JdbcStatementBuilder[T] {
      override def accept(statement: PreparedStatement, data: T): Unit = {
        val fields = data.getClass.getDeclaredFields
        var index = 1
        for (i <- 0 until fields.length) {
          breakable {
            val field = fields(i)
            val transientSink = field.getAnnotation(classOf[TransientSink])
            if (transientSink != null) {
              break()
            }
            field.setAccessible(true)

            statement.setObject(index, field.get(data))
            index = index + 1
          }
        }
      }
    }

    //设置执行参数
    val executionOptions = new JdbcExecutionOptions.Builder()
      .withBatchSize(2)
      .build()

    //设置clickhouse连接参数
    val connectOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
      //driver
      .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
      //url
      .withUrl(GmallConfig.CLICKHOUSE_URL)
      .build()
    JdbcSink.sink(sql, jdbcStatementBuilder, executionOptions, connectOptions)
  }

  /**
   * 需要实体类的字段和表的字段一直，且不需要INSERT的字段需要使用@TransientSink注解
   *
   * @param tableName
   * @param clazz
   * @tparam T
   * @return
   */
  def getSinkFunction[T](tableName: String, clazz: Class[T]) = {

    val sqlBuilder = new StringBuilder(s"INSERT INTO $tableName (")
    val fields = clazz.getDeclaredFields
    val buffer = new util.ArrayList[String]()
    val buffer2 = new util.ArrayList[String]()
    for (i <- 0 until fields.length) {
      breakable {
        val field = fields(i)
        val transientSink = field.getAnnotation(classOf[TransientSink])
        if (transientSink != null) {
          break()
        }
        field.setAccessible(true)
        buffer.add(field.getName)
        buffer2.add("?")
      }
    }
    val sql = sqlBuilder.append(StrUtil.join(", ", buffer))
      .append(s") VALUES(${StrUtil.join(", ", buffer2)})").toString()
    println(sql)

    //参数赋值
    val jdbcStatementBuilder = new JdbcStatementBuilder[T] {
      override def accept(statement: PreparedStatement, data: T): Unit = {
        val fields = clazz.getDeclaredFields
        var index = 1
        for (i <- 0 until fields.length) {
          breakable {
            val field = fields(i)
            val transientSink = field.getAnnotation(classOf[TransientSink])
            if (transientSink != null) {
              break()
            }
            field.setAccessible(true)
            statement.setObject(index, field.get(data))
            index = index + 1
          }
        }
      }
    }

    //设置执行参数
    val executionOptions = new JdbcExecutionOptions.Builder()
      .withBatchSize(2)
      .build()

    //设置clickhouse连接参数
    val connectOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
      //driver
      .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
      //url
      .withUrl(GmallConfig.CLICKHOUSE_URL)
      .build()
    JdbcSink.sink(sql, jdbcStatementBuilder, executionOptions, connectOptions)
  }

  def main(args: Array[String]): Unit = {
    getSinkFunction("product_stats", classOf[ProductStats])
  }

}
